package com.jfinal.plugin.avtivemq;

import java.util.ArrayList;
import java.util.List;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;

import com.jfinal.kit.StrKit;
import com.jfinal.plugin.IPlugin;

/**
 * @createTime: 2018/12/20
 * @author: HingLo
 * @description: activemq
 */
public class ActiveMqPlugin implements IPlugin {

	/**
	 * session 连接池默认大小,默认为队列数目的2倍+extraSession, 因为每个队列都需一个session来监听队列.该资源无法释放
	 * 同时还需要对每个队列进行添加元素也需要session, 设置extraSession
	 * 是额外的多创建部分session,用于高并发时,快速得到session
	 */
	private int extraSession = 2;
	/***
	 * 设置session连接池为消息队列的几倍，避免高并发发送消息因session不够而创建过多session对象，
	 * 因为session对象是重复利用的，并未进行释放，从而导致连接池中的session对象过多，而占用过多系统资源
	 * 例如：有100个消息队列，那么在连接池中最多也只能有1000*10个session，因为并发需要创建的session会自动释放链接
	 */
	private int max = 10;
	/***
	 * 是否支持事物
	 */
	private static boolean transacted = false;
	/**
	 * 事物提交类型
	 */
	private static int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
	/**
	 * 链接地址
	 */
	private String url;
	/**
	 * 用户名
	 */
	private String username;
	/***
	 * 密码
	 */
	private String password;
	/***
	 * 队列消息监听集合(一般有多少个队列/主题,就有多少个元素)
	 */
	private List<ConsumerMsg> consumerMsgList = new ArrayList<>();

	/***
	 * 扫描监听器的包目录,例如:com.hinglo,com, 等
	 */
	private String scanPackage = "com.lhm.center.mq";
	/**
	 * 是否启用注解扫描
	 */
	private boolean scan = false;

	public ActiveMqPlugin(String url) {
		this.url = url;
	}

	public ActiveMqPlugin(String url, String username, String password) {
		this(url);
		this.username = username;
		this.password = password;
	}

	/***
	 * 初始化连接池信息(session 连接池初始化)
	 * 
	 * @return 返回结果
	 */
	private boolean initConnection() {
		// 注解扫描初始化
		if (scan) {
			scan(this.scanPackage);
		}
		ConnectFactory.setSize(this.consumerMsgList.size() * 2 + extraSession);
		ConnectFactory.setAcknowledgeMode(acknowledgeMode);
		ConnectFactory.setTransacted(transacted);
		// 设置连接池最大的数量
		ConnectFactory.setMax(this.consumerMsgList.size() * max);
		try {
			if (StrKit.notBlank(username, password)) {
				ConnectFactory.init(url, username, password);
			} else {
				ConnectFactory.init(url);
			}
			// 初始化消费者监听
			return initConsumerMsg();
		} catch (JMSException e) {
			e.printStackTrace();
			return false;
		}
	}

	/**
	 * 绑定队列监听器
	 *
	 * @return 绑定结果
	 */
	private boolean initConsumerMsg() {
		consumerMsgList.forEach(x -> {
			try {
				Session session = SessionKit.getJmsSession();
				Destination destination = DestinationKit.createDestination(x.getType(), x.getName(), session);
				MessageConsumer messageConsumer = DestinationKit.createMessageConsumer(session, destination);
				messageConsumer.setMessageListener(x.getMessageListener());
			} catch (InterruptedException | JMSException e) {
				e.printStackTrace();
			}
		});
		return true;

	}

	/***
	 * 手动添加队列监听器信息
	 * 
	 * @param name          名称
	 * @param type          类型
	 * @param listenerClass 监听类
	 */
	public void addListener(String name, MessageType type, Class<? extends MessageListener> listenerClass) {
		ConsumerMsg consumerMsg = new ConsumerMsg();
		consumerMsg.setName(name);
		consumerMsg.setType(type);
		try {
			consumerMsg.setMessageListener(listenerClass.newInstance());
			this.consumerMsgList.add(consumerMsg);
		} catch (InstantiationException | IllegalAccessException e) {
			e.printStackTrace();
		}
	}

	/***
	 * 注解扫描监听器
	 * 
	 * @param scanPackage 定义的包
	 */
	public void scan(String scanPackage) {
		// 第一步,将获取包下面所有的class
		// 第二步,草扫描带有MsgListener注解的类
		for (Class<?> className : new ClassScanner(scanPackage).scan()) {
			try {
				MsgListener msgListener = className.getAnnotation(MsgListener.class);
				if (msgListener != null) {
					MessageListener messageListener = (MessageListener) className.newInstance();
					// 将扫描到的监听类放入到集合中
					for (String name : msgListener.value()) {
						ConsumerMsg consumerMsg = new ConsumerMsg();
						consumerMsg.setName(name);
						consumerMsg.setType(msgListener.type());
						consumerMsg.setMessageListener(messageListener);
						this.consumerMsgList.add(consumerMsg);
					}
				}
			} catch (IllegalAccessException | InstantiationException e) {
				e.printStackTrace();
			}
		}

	}

	@Override
	public boolean start() {
		return initConnection();
	}

	@Override
	public boolean stop() {
		return true;
	}

	public static void setTransacted(boolean transacted) {
		ActiveMqPlugin.transacted = transacted;
	}

	public static void setAcknowledgeMode(int acknowledgeMode) {
		ActiveMqPlugin.acknowledgeMode = acknowledgeMode;
	}

	public ActiveMqPlugin setExtraSession(int extraSession) {
		this.extraSession = extraSession;
		return this;
	}

	/***
	 * 是否启用注解扫描监听器
	 * 
	 * @param scan true
	 */
	public ActiveMqPlugin setScan(boolean scan) {
		this.scan = scan;
		return this;
	}

	/**
	 * 设置扫描监听器的包,
	 *
	 * @param scanPackage 包名称
	 */
	public ActiveMqPlugin setScanPackage(String scanPackage) {
		this.scanPackage = scanPackage;
		return this;
	}

	public ActiveMqPlugin setMax(int max) {
		this.max = max;
		return this;
	}
}
